package com.example.bookservice.kafka;

import com.example.commom.es.ElasticsearchDeleteCommand;
import com.example.commom.global.KafkaConstants;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.retry.annotation.Retryable;
import org.springframework.transaction.annotation.Transactional;

/**
 *  同步删除 ES
 */
@Slf4j
@Configuration
@Transactional
public class KafkaDeleteES {

    @Autowired
    private RestHighLevelClient client;

    @KafkaListener(topics = KafkaConstants.KAFKA_BOOK_DELETE_ES, groupId = KafkaConstants.KAFKA_GROUP_ID)
    public void deleteBookEs(String bookId,  Acknowledgment acknowledgment) {
        log.info("start delete book info, it's id is " + bookId);

        // 1. 准备 request
        DeleteRequest request = new DeleteRequest("book").id(bookId);

        // 准备 netflix
        ElasticsearchDeleteCommand command = new ElasticsearchDeleteCommand(client, request);

        // 发起请求
        Boolean delete = command.execute();

        if (delete) {
            acknowledgment.acknowledge();
        } else {
            log.error("failed");
            throw new RuntimeException("Simulating retriable exception");
        }

    }

    @KafkaListener(topics = KafkaConstants.KAFKA_CHAPTER_DELETE_ES, groupId = KafkaConstants.KAFKA_GROUP_ID)
    public void deleteChapterEs(String bookId,  Acknowledgment acknowledgment) {
        log.info("start delete chapter info, it's id is " + bookId);

        // 1. 准备 request
        DeleteRequest request = new DeleteRequest("chapter").id(bookId);

        // 准备 netflix
        ElasticsearchDeleteCommand command = new ElasticsearchDeleteCommand(client, request);

        // 发起请求
        Boolean delete = command.execute();

        if (delete) {
            log.error("failed");
            throw new RuntimeException("Simulating retriable exception");
        }
    }

}
